package main.java.hotProd

import java.util.Date

import cn.hutool.core.date.{DatePattern, DateUtil}
import org.apache.log4j.{Level, Logger}
import org.apache.spark.sql.SparkSession



/**
  * HotProdSparkInArea
  * 各地区商品热度查询的SPARK
  * @author zhangyimin
  * 2018-11-12 下午4:32
  * @version 1.0
  */
object HotProdSparkInArea {

  def main(args: Array[String]): Unit = {

    //屏蔽日志
    Logger.getLogger("org.apache.spark").setLevel(Level.ERROR)
    Logger.getLogger("org.eclipse.jetty.server").setLevel(Level.OFF)
    val sparkSession = SparkSession.builder().master("local").appName("HotProdInArea").getOrCreate()

    val sc = sparkSession.sparkContext

    val clickLogRdd = sc
      .textFile("hdfs://10.16.7.36:9000/user/hive/warehouse/clicklog")
      .map(_.split(","))
      .map(x => {
        new clickLog(x(2), x(5))
      })
    val areaRdd = sc
      .textFile("hdfs://10.16.7.36:9000/data/input/hot_products/area")
      .map(_.split(","))
      .map(x => {
        new area(x(0), x(1))
      })
    val prodRdd = sc
      .textFile("hdfs://10.16.7.36:9000/data/input/hot_products/product")
      .map(_.split(","))
      .map(x => {
        new product(x(0), x(2))
      })


    import sparkSession.implicits._

    clickLogRdd.toDF().createOrReplaceTempView("clickLog")
    areaRdd.toDF().createOrReplaceTempView("area")
    prodRdd.toDF().createOrReplaceTempView("product")


//spark解析URL
//    import scala.sys.process.processInternal.URL
//    URLUtil.toURI("").
//    urls.getQuery()



    val a=sparkSession.sql(" select\n  a.area_id,\n  a.area_name,\n  c.product_id,\n  c.product_name,\n  count(b.product_id) clickCount\nfrom area a, clicklog b, product c\nwhere a.area_id = b.area_id and b.product_id = c.product_id\ngroup by a.area_id,\n  a.area_name,\n  c.product_id,\n  c.product_name order by clickCount")

//    val b=sparkSession.sql(" select\n  a.area_id,\n  a.area_name,\n  c.product_id,\n  c.product_name,\n  count(b.product_id)\nfrom clicklog b\n  join area a on b.area_id = a.area_id\n  join product c on b.product_id = c.product_id\ngroup by a.area_id,\n  a.area_name,\n  c.product_id,\n  c.product_name")
//
//    val c= sparkSession.sql(" select\n  a.area_id,\n  a.area_name,\n  c.product_id,\n  c.product_name,\n  count(b.product_id)\nfrom clicklog b\n  left join area a on b.area_id = a.area_id\n  left join product c on b.product_id = c.product_id\ngroup by a.area_id,\n  a.area_name,\n  c.product_id,\n  c.product_name")


    a.rdd.saveAsTextFile("hdfs://10.16.7.36:9000/data/output/explain/area"+DateUtil.format(new Date(),DatePattern.PURE_DATETIME_MS_PATTERN))


    sparkSession.stop()







    //    clickLogRdd.map((_(2),_(5)))


  }


  case class area(area_id: String, area_name: String)
  case class clickLog(product_id: String, area_id: String)
  case class product(product_id: String, product_name: String)




}
